package com.darkidiot.session.serialize;

import com.darkidiot.session.exception.SerializableException;
import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Input;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.serializers.JavaSerializer;
import com.esotericsoftware.kryo.serializers.MapSerializer;
import com.google.common.base.Throwables;
import com.google.common.io.BaseEncoding;
import lombok.extern.slf4j.Slf4j;

import java.io.ByteArrayInputStream;
import java.io.ByteArrayOutputStream;
import java.io.IOException;
import java.util.HashMap;
import java.util.Map;

/**
 * Kryo序列化支持类<>速度快，但是只支持HashMap</>
 * session attribute持久化到redis
 * Copyright (c) for darkidiot
 * Date:2017/4/14
 * Author: <a href="darkidiot@icloud.com">darkidiot</a>
 * School: CUIT
 * Desc:
 */
@Slf4j
public class KryoSerializer implements Serializer {

    private ThreadLocal<Kryo> kryoSerializeLocal = new ThreadLocal<Kryo>() {
        @Override
        protected Kryo initialValue() {
            Kryo kryoSerialize = new Kryo();
            kryoSerialize.setReferences(false);
            kryoSerialize.setRegistrationRequired(true);
            kryoSerialize.register(HashMap.class, serializerLocal.get());
            return kryoSerialize;
        }

    };

    private ThreadLocal<MapSerializer> serializerLocal = new ThreadLocal<MapSerializer>() {
        @Override
        protected MapSerializer initialValue() {

            MapSerializer serializer = new MapSerializer();
            serializer.setKeyClass(String.class, new JavaSerializer());
            serializer.setKeysCanBeNull(false);
            serializer.setValueClass(Object.class, new JavaSerializer());
            serializer.setValuesCanBeNull(true);
            return serializer;
        }
    };


    @Override
    public String serialize(Map<String, Object> o) {
        return serializationMap(o);
    }

    @Override
    public Map<String, Object> deserialize(String o) throws SerializableException {
        return deserializationMap(o);
    }

    private String serializationMap(Map<String, Object> obj) {
        try (ByteArrayOutputStream byteArrayOutputStream = new ByteArrayOutputStream()) {
            Output output = new Output(byteArrayOutputStream);
            kryoSerializeLocal.get().writeObject(output, obj);
            output.flush();
            output.close();
            byte[] b = byteArrayOutputStream.toByteArray();
            byteArrayOutputStream.flush();
            return BaseEncoding.base16().encode(b);
        } catch (IOException e) {
            log.error("failed to serialize http session {} to binary,cause:{}", obj, Throwables.getStackTraceAsString(e));
            throw new SerializableException(SerializableException.Code.SERIALIZE_ERROR, "failed to serialize http session to binary", e);
        }
    }

    @SuppressWarnings("unchecked")
    private Map<String, Object> deserializationMap(String obj) {
        try {
            ByteArrayInputStream byteArrayInputStream = new ByteArrayInputStream(BaseEncoding.base16().decode(obj));
            Input input = new Input(byteArrayInputStream);
            return kryoSerializeLocal.get().readObject(input, HashMap.class, serializerLocal.get());
        } catch (Exception e) {
            log.error("failed to deserialize string  {} to http session,cause:{} ", obj, Throwables.getStackTraceAsString(e));
            throw new SerializableException(SerializableException.Code.DESERIALIZE_ERROR, "failed to deserialize string to http session", e);
        }
    }
}
